package com.headfirst.dmp.tools

import ch.hsr.geohash.GeoHash
import com.headfirst.dmp.utils.{GaodeGeoAPI, RedisUtils}
import org.apache.commons.lang.StringUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import redis.clients.jedis.JedisCluster

/**
  * 抽取经纬度信息，将转化成商圈信息，保存到Redis中
  */
object ExtractLatLong2Busniess {


  def main(args: Array[String]): Unit = {

    //1.验证参数
    if (args.length != 1) {
      print(
        """
          |com.headfirst.dmp.tools.ExtractLatLong2Busniess
          |需要参数：
          |       loginputPath
          |
        """.stripMargin)
      sys.exit(-1)
    }

    //2.接收参数
    val Array(loginputPath) = args

    //3.创建SparkSession对象
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.setMaster("local[*]")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //设置序列化方式采用KryoSerializer方式（默认的是java序列化）

    val session: SparkSession = SparkSession.builder().config(conf).getOrCreate()


    //4.读取文件并处理
    // 通过高德API查找对应的商圈信息，然后保存到Redis中
    session.read.parquet(loginputPath)
      .select("lat", "longs")
      .where("lat >= 3.86 and lat <= 53.55  and longs >=73.66  and longs <= 135.05")
      .distinct()
      .foreachPartition(itr => {

        //获取CLUSTER操作连接对象
        val jedis: JedisCluster = RedisUtils.getRedisCluster()

        itr.foreach(row => {

          //获取两个字段
          val lat = row.getAs[String]("lat")
          val longs = row.getAs[String]("longs")

          //两个字段计算GeoHash
          val latDouble: Double = lat.toDouble
          val longsDouble: Double = longs.toDouble
          val base3: String = GeoHash.withCharacterPrecision(latDouble, longsDouble, 8).toBase32

          //获取API所有信息
         val businessName: String = GaodeGeoAPI.getBuuiness(longs + "," + lat)

          //存入Redis中
          if (StringUtils.isNotBlank(businessName)){
            jedis.set(base3,businessName)
          }

        })
        //关流,JedisCluster自动关流，不需要手动关流，如果手动关，会导致集群连接关闭！
        //jedis.close()
      })


    //关流
    session.stop()


  }

}
